package com.cml.getway.core.netty.processor;

import com.cml.getway.common.concurrent.queue.flusher.ParallelFlusher;
import com.cml.getway.common.enums.ResponseCode;
import com.cml.getway.core.GetwayConfig;
import com.cml.getway.core.context.HttpRequestWrapper;
import com.cml.getway.core.helper.ResponseHelper;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;

/**
 * 姓名：陈茂林
 * 时间：2023/12/20
 * 描述：
 */
@Slf4j
public class NettyBatchEventProcessor implements NettyProcessor{
    private GetwayConfig getwayConfig;
    private  NettyCoreProcessor nettyCoreProcessor;

    private ParallelFlusher<HttpRequestWrapper> parallelFlusher;

    private static final String THREAD_NAME_PREFIX = "getway-flusher-";

    public NettyBatchEventProcessor(GetwayConfig getwayConfig,NettyCoreProcessor nettyCoreProcessor){
        this.getwayConfig=getwayConfig;
        this.nettyCoreProcessor=nettyCoreProcessor;
        ParallelFlusher.Builder<HttpRequestWrapper> builder = new ParallelFlusher.Builder<HttpRequestWrapper>()
                .setBufferSize(getwayConfig.getBufferSize())
                .setThreads(getwayConfig.getProcessThread())
                .setProducerType(ProducerType.MULTI)
                .setNamePrefix(THREAD_NAME_PREFIX)
                .setWaitStrategy(getwayConfig.getATureWaitStrategy());
        BatchEventProcessorListener batchEventProcessorListener = new BatchEventProcessorListener();
        builder.setEventListener(batchEventProcessorListener);
        this.parallelFlusher = builder.build();
    }
    @Override
    public void process(HttpRequestWrapper httpRequestWrapper) {
        this.parallelFlusher.add(httpRequestWrapper);
    }

    @Override
    public void start() {
        this.nettyCoreProcessor.start();
        this.parallelFlusher.start();
    }

    @Override
    public void shutdown() {
        this.nettyCoreProcessor.shutdown();
        this.parallelFlusher.shutdown();
    }


    public class BatchEventProcessorListener implements ParallelFlusher.EventListener<HttpRequestWrapper> {

        @Override
        public void onEvent(HttpRequestWrapper event) throws Exception {
            nettyCoreProcessor.process(event);
        }

        @Override
        public void onException(Throwable t, long sequence, HttpRequestWrapper event) {
            HttpRequest request = event.getFullHttpRequest();
            ChannelHandlerContext ctx = event.getCtx();
            try {
                log.error("#BatchEventProcessorListener# onException 请求处理失败, request: {}. errorMessage: {}",
                        request, t.getMessage(), t);

                //	首先构建响应对象
                FullHttpResponse fullHttpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);
                //	判断是否保持连接
//                if(!HttpUtil.isKeepAlive(request)) {
                    ctx.writeAndFlush(fullHttpResponse).addListener(ChannelFutureListener.CLOSE);
//                } else {
                    //	如果保持连接, 则需要设置一下响应头：key: CONNECTION,  value: KEEP_ALIVE
//                    fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
//                    ctx.writeAndFlush(fullHttpResponse);
//                }

            } catch (Exception e) {
                //	ignore
                log.error("#BatchEventProcessorListener# onException 请求回写失败, request: {}. errorMessage: {}",
                        request, e.getMessage(), e);
            }
        }
    }
}
